package org.example.ItemCF.src.recommend;

import java.io.IOException;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.example.ItemCF.src.hdfs.HdfsDAO;


/**
 * 此类负责实现物品之间的局部乘法，然后对结果进行聚合，以生成最终的推荐列表。
 */
public class Step4Driver {

    /**
     * 提供一个方法来运行Step4的MapReduce作业。
     * @param path 包含输入和输出路径的映射
     * @throws IOException
     * @throws ClassNotFoundException
     * @throws InterruptedException
     */
    public static void run(Map<String, String> path) throws IOException,
            ClassNotFoundException, InterruptedException {

        Configuration conf = new Configuration();
        String input1 = path.get("Step4Input1");
        String input2 = path.get("Step4Input2");
        String output = path.get("Step4Output");

        HdfsDAO hdfs = new HdfsDAO(Recommend.HDFS, conf);
        hdfs.rmr(output);

        Job job = Job.getInstance(conf, "RecommendStep1");
        job.setJarByClass(Step4Driver.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        job.setMapperClass(Step4PartialMultiplyMapper.class);
        job.setCombinerClass(Step4AggregateAndRecommendReducer.class);
        job.setReducerClass(Step4AggregateAndRecommendReducer.class);
        FileInputFormat.setInputPaths(job, new Path(input1), new Path(input2));
        FileOutputFormat.setOutputPath(job, new Path(output));
        if (!job.waitForCompletion(true))
            return;

    }

}



